package com.nl.pulsardemo.pool.service.impl;

import com.nl.pulsardemo.pool.config.commons.utils.ResultVOUtils;
import com.nl.pulsardemo.pool.service.IConsumer;
import com.nl.pulsardemo.pool.vo.ResultVO;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.PulsarClientException;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

/**
 * @author : ZhaoY
 * @description :
 * @date : 2021-02-01 16:14
 **/
@Slf4j
@Service
public class ConsumerImpl implements IConsumer {
    static Integer index = 0;

    @Override
    public ResultVO SyncReceive(Consumer consumer) throws PulsarClientException {
        index++;
        Message<String> message = null;
        ResultVO resultVO;
        try {
//            message = consumer.receive(5, TimeUnit.MILLISECONDS);
            message = consumer.receive();
            if (message == null) {
                resultVO = ResultVOUtils.error(-1, "消息队列暂无数据可消费");
            } else {
                resultVO = ResultVOUtils.success((Object) message.getValue());
//                log.info(index + ":receive message: " + (Integer.valueOf(new String(message.getData())) - 1));
                log.info(":receive message: " + (Object) message.getValue());
                consumer.acknowledge(message);
            }
        } catch (Exception e) {
            log.info("消费数据失败");
            consumer.negativeAcknowledge(message);
            resultVO = ResultVOUtils.error(-1, "消费数据失败");
        }
        return resultVO;
    }

    @Override
    public ResultVO AsyncReceive(Consumer consumer) {
        ResultVO resultVO = null;
        CompletableFuture<Message> asyncMessage = consumer.receiveAsync();
        Message<String> message = null;
        try {
            message = asyncMessage.get();
            log.info(":receive message: " + (Object) message.getValue());
            consumer.acknowledge(message);
            resultVO = ResultVOUtils.success((Object) message.getValue());
        } catch (InterruptedException e) {
            log.info("消费数据失败");
            resultVO = ResultVOUtils.error(-1, "消费数据失败");
        } catch (ExecutionException e) {
            resultVO = ResultVOUtils.error(-1, "消费数据失败");
        } catch (PulsarClientException e) {
            consumer.negativeAcknowledge(message);
            resultVO = ResultVOUtils.error(-1, "消费数据失败");
        }

        return resultVO;
    }
}
